package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"net"
	custormconfig "socketTokafka/config"
	loggger "socketTokafka/logger"
	"strconv"
	"strings"
	"time"
)

//// 限制goroutine数量
//var limitChan = make(chan bool, 2000) // Todo 从配置文件中读取
//var connectTimes = 0
//
//// 限制同时处理消息数量
//var msgChan = make(chan string, 2000) // Todo 从配置文件中读取
//var kafkaLimitedChan = make(chan bool, 100)
//var kafkaProduct sarama.SyncProducer
//var udpAddress = "0.0.0.0" // Todo 从配置文件中读取
//var udpPort = "7777"       // Todo 从配置文件中读取
// UDP goroutine 实现并发读取UDP数据

var kafkaConfig = custormconfig.InitKafkaConfig()
var udpConfig = custormconfig.InitUdpConfig()
var threadConfig = custormconfig.InitThreadConfig()
var limitChan = make(chan bool, threadConfig.LimitChan)
var msgChan = make(chan string, threadConfig.MsgChanLimit) // Todo 从配置文件中读取
var kafkaLimitedChan = make(chan bool, threadConfig.KafkaLimitedChan)
var kafkaProduct sarama.SyncProducer
var connectTimes = 0
var topica = kafkaConfig.TopicA
var topicb = kafkaConfig.TopicB
var topicc = kafkaConfig.TopicC

func udpProcess(conn *net.UDPConn) {
	defer func() {
		// 释放出一个协程
		<-limitChan
	}()

	// 最大读取数据大小
	data := make([]byte, 32767)
	n, _, err := conn.ReadFromUDP(data)
	if err != nil {
		panic(err)
	}
	// 获取对端的IP地址
	// remoteAddr := conn.RemoteAddr()
	// msgChan <- remoteAddr.String() + " " + string(data[:n])
	msgChan <- string(data[:n])
}

func udpServer(address, port string) {
	// @todo 如何防止udpServer 一直Panic导致无限循环重启
	defer func() {
		if e := recover(); e != nil {
			// udpServer启动失败后，间隔10秒后重试
			time.Sleep(10 * time.Second)
			udpServer(udpConfig.UdpAddress, udpConfig.UdpPort)
		}
	}()
	udpAddr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(address, port))
	conn, err := net.ListenUDP("udp", udpAddr)
	loggger.Logger.Info("监听[UDP]地址：%v, 端口：%v", udpConfig.UdpAddress, udpConfig.UdpPort)
	defer conn.Close()
	if err != nil {
		panic(err)
	}

	for {
		limitChan <- true
		go udpProcess(conn)
	}
}

func sendMessage(c chan string) {
	// @todo 如何防止activeMqProducer 退出
	defer func() {
		// 重试
		go sendMessage(msgChan)
	}()
	for {
		kafkaLimitedChan <- true // 限制开启协程数量
		contentMsg := <-c
		go func() {
			defer func() {
				// 释放出一个协程
				<-kafkaLimitedChan
			}()
			var msg *sarama.ProducerMessage
			if find := strings.Contains(contentMsg, "\"msgtype\": 200"); find {
				loggger.Logger.Info("productA recive message info:%v.", contentMsg)
				loggger.Logger.Info("productA find the character.")
				msg = &sarama.ProducerMessage{
					Topic:     topica,
					Partition: int32(-1),
					Key:       sarama.StringEncoder(""),
					Value:     sarama.ByteEncoder(contentMsg),
				}
			} else if findc := strings.Contains(contentMsg, "\"msgtype\": 201"); findc {
				loggger.Logger.Info("productC find the character.")
				msg = &sarama.ProducerMessage{
					Topic:     topicc,
					Partition: int32(-1),
					Key:       sarama.StringEncoder(""),
					Value:     sarama.ByteEncoder(contentMsg),
				}
			} else {
				msg = &sarama.ProducerMessage{
					Topic:     topicb,
					Partition: int32(-1),
					Key:       sarama.StringEncoder(""),
					Value:     sarama.ByteEncoder(contentMsg),
				}
			}
			paritition, offset, err := kafkaProduct.SendMessage(msg)
			//loggger.Logger.Info("Partion = %v, offset = %v", paritition, offset)
			if err != nil {
				fmt.Println("Send Message Fail")
				panic(err)
				loggger.Logger.Error("Partion = %v, offset = %v", paritition, offset)
			}
		}()
	}
}

// 使用IP和端口连接到kafka服务器, 返回kafka连接对象
func connKafkaProduct() {
	// @todo 如何防止无限循环
	defer func() {
		if e := recover(); e != nil {
			// 记录错误日志
			loggger.Logger.Error("connActiveMq error:%v", e)
			time.Sleep(3 * time.Second)
			kafkaProduct = nil
			connKafkaProduct()
		}
	}()
	// @todo 实现断开重连
	if kafkaProduct == nil {
		var err error
		config := sarama.NewConfig()
		config.Producer.Return.Successes = true
		config.Producer.RequiredAcks = sarama.WaitForAll
		config.Producer.Partitioner = sarama.NewRandomPartitioner
		kafkaProduct, err = sarama.NewSyncProducer(strings.Split(kafkaConfig.Broker, ","), config)
		if err != nil {
			var err error
			connectTimes++
			if connectTimes >= 100 {
				time.Sleep(60 * time.Second)
			} else if connectTimes >= 10 {
				time.Sleep(10 * time.Second)
			} else {
				time.Sleep(3 * time.Second)
			}
			panic(err.Error() + ", 重新连接ActiveMQ, 已重试次数: " + strconv.Itoa(connectTimes))

		} else {
			connectTimes = 0
		}
	}
}

func init() {
	connKafkaProduct()
	go sendMessage(msgChan)
}

func main() {
	udpServer(udpConfig.UdpAddress, udpConfig.UdpPort)
}
